package org.lzfto.flink.demo.controller;

import org.apache.commons.lang3.StringUtils;
import org.lzfto.flink.demo.business.flinkManager.FlinkManageService;
import org.lzfto.flink.demo.business.flinkManager.JarJobParam;
import org.lzfto.flink.demo.common.result.PubResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.multipart.MultipartFile;

/**
 * 处理Flink的作业和jar
 */
@RestController
@RequestMapping("/flink/")
public class FlinkController {

	@Autowired
	private FlinkManageService flinkManageService;

	@GetMapping("/job/getJobs")
	public PubResult<String> getJobs(@RequestParam(value = "url", required = false) String url) {
		return flinkManageService.getJobs(validateUrl(url));
	}

	@GetMapping("/job/getJobStatus/{jobId}")
	public PubResult<String> getJobStatus(@RequestParam(value = "url", required = false) String url,
			@PathVariable String jobId) {
		// TODO::获取job状态，并返回
		return PubResult.ok_data(null);
	}

	@GetMapping("/job/getJobResult/{jobId}")
	public PubResult<String> getJobResult(@RequestParam(value = "url", required = false) String url,
			@PathVariable String jobId) {
		// TODO::获取job结果，并返回
		return PubResult.ok_data(null);
	}

	@GetMapping("/job/cancelJob/{jobId}")
	public PubResult<Boolean> cancelJob(@PathVariable String jobId,
			@RequestParam(value = "url", required = false) String url) {
		return flinkManageService.cancelJob(jobId, validateUrl(url));
	}

	@PostMapping("/jars/upload")
	public PubResult<String> uploadJar(MultipartFile file,
			@RequestParam(value = "url", required = false) String url) {
		if (file.isEmpty()) {
			return PubResult.error("上传文件不能为空");
		}
		if (!file.getOriginalFilename().endsWith(".jar")) {
			return PubResult.error("只支持上传JAR文件");
		}
		return flinkManageService.uploadJar(file,url);
	}

	@GetMapping("/jars/getJars")
	public PubResult<String> getJars(@RequestParam(value = "url", required = false) String url) {
		return flinkManageService.getJars(validateUrl(url));
	}

	@PostMapping("/jars/run")
	public PubResult<String> run(@RequestBody JarJobParam param) {
		return flinkManageService.runJar(param);
	}

	@GetMapping("/jars/delete/{jarId}")
	public PubResult<Boolean> deleteJar(@PathVariable String jarId,
			@RequestParam(value = "url", required = false) String url) {
		return flinkManageService.deleteJar(jarId, validateUrl(url));
	}

	private String validateUrl(String url) {
		if (StringUtils.isEmpty(url)) {
			return url;
		} else {
			return url.endsWith("/") ? url : url + "/";
		}
	}

}
